package com.atguigu.realtime.app.dws;

import com.atguigu.realtime.app.BaseSqlApp;
import com.atguigu.realtime.bean.KeywordStats;
import com.atguigu.realtime.function.IkAnalyzer;
import com.atguigu.realtime.function.KwProduct;
import com.atguigu.realtime.util.FlinkSinkUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneOffset;

import static com.atguigu.realtime.common.Constant.*;


public class DwsProductKeywordStatsApp extends BaseSqlApp {
    public static void main(String[] args) {
        new DwsProductKeywordStatsApp().init(4005,1,"DwsProductKeywordStatsApp");
    }
    @Override
    protected void run(StreamTableEnvironment tEnv) {
          tEnv.getConfig().setLocalTimeZone(ZoneOffset.ofHours(8));

        //1.建立动态表 从kafka读数据:dwd_page
        tEnv.executeSql("create table product_stats(" +
                " stt string, " +
                " edt string, " +
                " sku_name string, " +
                " click_ct bigint, " +
                " order_ct bigint, " +
                " cart_ct bigint " +
                ")with(" +
                " 'connector'='kafka'," +
                " 'properties.bootstrap.servers'='" + KAFKA_BROKERS + "'," +
                " 'properties.group.id'='DwsProductKeywordStatsApp'," +
                " 'topic'='" + TOPIC_DWS_PRODUCT_STATS + "'," +
                " 'format'='json'," +
                " 'scan.startup.mode'='earliest-offset' " +
                ")");
        //tEnv.sqlQuery("select * from product_stats").execute().print();
        //1.过滤出需要得数据: 三个ct至少一个不为0
        Table t1 = tEnv.sqlQuery("select " +
                " * " +
                " from " +
                " product_stats " +
                " where click_ct >0 " +
                " or order_ct > 0 " +
                " or cart_ct >0 ");
        tEnv.createTemporaryView("t1",t1);


        // 2. 分词
        tEnv.createTemporaryFunction("ik_analyzer",IkAnalyzer.class);
        Table t2 = tEnv.sqlQuery("select " +
                " stt," +
                " edt," +
                " keyword," +
                " click_ct," +
                " order_ct," +
                " cart_ct " +
                " from t1 " +
                " join lateral table(ik_analyzer(sku_name)) on true ");
        tEnv.createTemporaryView("t2",t2);

        //3.使用表值函数,列转行
        tEnv.createTemporaryFunction("kw_product", KwProduct.class);
        Table t3 = tEnv.sqlQuery("select " +
                " stt, " +
                " edt, " +
                " keyword ," +
                " source," +
                " ct," +
                " unix_timestamp() * 1000 ts " +
                " from t2 " +
                " join lateral table(kw_product(click_ct,order_ct,cart_ct)) on true ");
        tEnv.createTemporaryView("t3",t3);
             // 4.聚合: 按照stt,edt,keyword,source,ct聚合
        Table result = tEnv.sqlQuery("select " +
                " stt, " +
                " edt, " +
                " keyword ," +
                " source," +
                " sum(ct) ct," +
                " unix_timestamp() * 1000 ts " +
                " from t3 " +
                " group by " +
                " stt, " +
                " edt, " +
                " keyword," +
                " source ");
        tEnv.toRetractStream(result,KeywordStats.class)
                .filter(t ->t.f0)
                .map(t->t.f1)
                .addSink(FlinkSinkUtil.getClickHouseSink(
                        "gmall2021","keyword_stats_2021",KeywordStats.class));
    }
}
